package com.scheduler.spark;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.springframework.stereotype.Service;
import scala.Tuple2;
import java.io.Serializable;
import java.util.*;

@Service
public class HopSearchWords implements Serializable {
/*    @Autowired
    private JavaSparkContext sc;*/

    public List<Tuple2<String, Integer>> WordCount(List<String> list){
        //配置执行
        SparkConf conf = new SparkConf().setAppName("WordCount").setMaster("local[1]") ;
        JavaSparkContext sc = new JavaSparkContext(conf);
        for (int j=0;j<list.size();j++){
            System.out.println(list.get(j)+"----------------------------------------");
        }
        JavaRDD<String> rddSource = sc.parallelize(list);
        JavaPairRDD<String,Integer> pairrdd = rddSource.mapToPair(new PairFunction<String, String, Integer>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<String, Integer> call(String word) throws Exception {
                return new Tuple2<String,Integer>(word,1);
            }
        });
        JavaPairRDD<String, Integer> count = pairrdd.reduceByKey(new Function2<Integer, Integer, Integer>() {
            private static final long serialVersionUID = 2L;
            @Override
            public Integer call(Integer i, Integer j) throws Exception {
                return i+j;
            }
        });
        List<Tuple2<String, Integer>> toplist = count.collect();
        Map<String, Integer> map = new HashMap<String , Integer> ();
         for (int i =0;i<toplist.size();i++){
            map.put(toplist.get(i)._1,toplist.get(i)._2);
        }
        //根据结果的键值对, 按值排序
        List<Map.Entry<String, Integer>> sortedWord = new ArrayList<Map.Entry<String, Integer>>(map.entrySet());
        Collections.sort(sortedWord, new Comparator<Map.Entry<String, Integer>>(){
            public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2){
                return (o2.getValue() - o1.getValue());
            }
        });
        List<Map.Entry<String, Integer>> entries = sortedWord.subList(0, sortedWord.size()<=3?sortedWord.size():3);
        List<Tuple2<String, Integer>> lt = new ArrayList<Tuple2<String, Integer>>();
        for(Map.Entry<String, Integer> word : entries){
            lt.add(new Tuple2<String,Integer>(word.getKey(),word.getValue()));
            System.out.println(word.getKey() + "   " + word.getValue());
        }
        sc.close();
        return lt;
    }
}
